package com.atguigu.rabbitmq.four;

import com.atguigu.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/*
* 生产方的发布确认模式：分别比较他们使用的时间，来比较那种确认方式是最好的
* 1.单个确认模式 2585ms
* 2.批量确认模式 208ms
* 3.异步批量确认模式 61ms
* */
public class ConfirmMessage {

    /*批量发送消息的个数*/
    public static final int MESSAGE_COUNT=1000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        /*1.调取单个确认模式的相关方法*/
//        ConfirmMessage.publishMessageIndividually();
        /*2.调取批量确认模式的相关方法*/
//        ConfirmMessage.publishMessageBatch();
        /*3.调取异步批量确认模式的相关方法*/
        ConfirmMessage.publishMessageAsync();
    }

    /*单个确认模式*/
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        /*得到连接里面的一个信道*/
        Channel channel = RabbitMqUtils.getChannel();
        /*队列的声明*/
        String queueName = UUID.randomUUID().toString();
        /*
        * 1.队列的名称
        * 2.队列里面的消息是否持久化
        * 3.队列里面的消息是否让多个消费者进行共享
        * 4.是否自动删除
        * 5.其他参数，关于延迟消息，死信队列这些
        * */
        channel.queueDeclare(queueName, true, false, false, null);
        /*开启发布确认*/
        channel.confirmSelect();
        /*开始时间*/
        long begin = System.currentTimeMillis();

        /*批量的发送消息*/
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            /*1.交换机名称 2.队列名称 3.其他参数，比如消息持久化等 4.消息的二进制*/
            channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
            /*单个消息确认*/
            boolean flag = channel.waitForConfirms();
//            if(flag){
//                System.out.println("消息发送成功");
//            }
        }

        /*结束时间*/
        long end=System.currentTimeMillis();

        System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息，耗时"+(end-begin)+"ms");
    }

    /*批量确认模式*/
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        /*得到连接里面的一个信道*/
        Channel channel = RabbitMqUtils.getChannel();
        /*队列的声明*/
        String queueName = UUID.randomUUID().toString();
        /*
         * 1.队列的名称
         * 2.队列里面的消息是否持久化
         * 3.队列里面的消息是否让多个消费者进行共享
         * 4.是否自动删除
         * 5.其他参数，关于延迟消息，死信队列这些
         * */
        channel.queueDeclare(queueName, true, false, false, null);
        /*开启发布确认*/
        channel.confirmSelect();
        /*开始时间*/
        long begin = System.currentTimeMillis();

        /*批量确认消息大小*/
        int batchSize=100;
        /*下面演示批量发送消息，并且批量发布确认*/
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            /*1.交换机名称 2.队列名称 3.其他属性 4.消息的二进制*/
            channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            /*判断达到100条消息的时候，批量的确认一次*/
            if(i%batchSize==0){
                /*发布确认消息*/
                boolean falg = channel.waitForConfirms();
            }
        }

        /*结束时间*/
        long end=System.currentTimeMillis();

        System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息，耗时"+(end-begin)+"ms");

    }


    /*异步批量确认模式*/
    public static void publishMessageAsync() throws IOException, TimeoutException {
        /*得到连接里面的一个信道*/
        Channel channel = RabbitMqUtils.getChannel();
        /*队列的声明*/
        String queueName = UUID.randomUUID().toString();
        /*
         * 1.队列的名称
         * 2.队列里面的消息是否持久化
         * 3.队列里面的消息是否让多个消费者进行共享
         * 4.是否自动删除
         * 5.其他参数，关于延迟消息，死信队列这些
         * */
        channel.queueDeclare(queueName, true, false, false, null);
        /*开启发布确认*/
        channel.confirmSelect();

        /*
        * 线程安全有序的一个hash表，适用于高并发的情况下
        * 1.轻松的将序号与消息进行关联
        * 2.可以轻松批量删除条数，只要给到序号
        * 3.支持高并发(多线程)
        * */
        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();

        /*开始时间*/
        long begin = System.currentTimeMillis();


        /*发送消息成功的方法*/
        /*
        * 参数说明
        * 1.消息的标记
        * 2.是否为批量确认
        * */
        ConfirmCallback ackCallback=(deliveryTag,multiple)->{
            if(multiple){
                /*1..2:删除已经确认的消息，剩下的就是没有确认的消息*/
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            }else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息:"+deliveryTag);
        };
        /*发送消息失败的方法*/
        ConfirmCallback nackCallback=(deliveryTag,multiple)->{
            /*1..3:打印下未确认的消息都有哪些*/
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息是:"+message+"::::未确认的消息tag:"+deliveryTag);
        };
        /*准备消息的监听器，监听那些消息成功了，那些消息失败*/
        /*两个参数的监听器，监听发送消息成功了的方法，监听发送消息失败了的方法*/
        /*这里的监听是异步的,当把消息发送出去之后，这个监听器线程才会触发 */
        channel.addConfirmListener(ackCallback,nackCallback);

        /*批量发送消息，异步确认消息*/
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            /*1..1:此处记录下所有要发送的消息，消息的总和*/
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }

        /*结束时间*/
        long end=System.currentTimeMillis();

        System.out.println("发布"+MESSAGE_COUNT+"个异步批量确认消息，耗时"+(end-begin)+"ms");

    }


}